package com.alipay.sofa.jraft.example.rheakv;

import com.alipay.sofa.jraft.Closure;
import com.alipay.sofa.jraft.Status;
import com.alipay.sofa.jraft.core.NodeImpl;
import com.alipay.sofa.jraft.util.DisruptorBuilder;
import com.alipay.sofa.jraft.util.NamedThreadFactory;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.jetbrains.annotations.NotNull;

import java.util.Random;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author chy
 * @description
 * @date 2024/12/14
 */
public class Main {
    public static void main(String[] args) throws InterruptedException {
        Disruptor<DemoClosure> applyDisruptor = DisruptorBuilder.<DemoClosure> newInstance() //
                .setRingBufferSize(1024) //
                .setEventFactory(new DemoClosureFactory()) //
                .setThreadFactory(new DemoThreadFactory()) //
                .setProducerType(ProducerType.MULTI) //
                .setWaitStrategy(new BlockingWaitStrategy()) //
                .build();
        applyDisruptor.handleEventsWith(new DemoHandler());
        RingBuffer<DemoClosure> applyQueue = applyDisruptor.start();
        for (int i = 0; i < 5; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    Random random = new Random();

                    while (true) {
                        int sleep = random.nextInt(10);
                        try {
                            Thread.sleep(sleep);
                            applyQueue.publishEvent(new EventTranslator<DemoClosure>() {
                                @Override
                                public void translateTo(DemoClosure event, long sequence) {
                                }
                            });
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                }
            }).start();
        }

        Thread.currentThread().join();
    }

    private static class DemoHandler implements EventHandler<DemoClosure> {
        @Override
        public void onEvent(DemoClosure event, long sequence, boolean endOfBatch) throws Exception {
            System.out.println(endOfBatch);
            event.run(Status.OK());
        }
    }

    private static class DemoThreadFactory implements ThreadFactory {
        private static final AtomicInteger ID = new AtomicInteger(0);
        @Override
        public Thread newThread(@NotNull Runnable r) {
            return new Thread(r, "test-" + ID.getAndIncrement());
        }
    }

    private static class DemoClosure implements Closure {
        @Override
        public void run(Status status) {
            System.out.println(Thread.currentThread().getName() + ": close closure");
        }
    }

    private static class DemoClosureFactory implements EventFactory<DemoClosure> {

        @Override
        public DemoClosure newInstance() {
            return new DemoClosure();
        }
    }

}
